#pragma once

//#include <srs_app_recv_thread.hpp>
#include <srs_core.hpp>
#include <srs_app_thread.hpp>
#include <srs_protocol_buffer.hpp>
#include <srs_core_performance.hpp>
#include <srs_app_reload.hpp>

#include <vector>

class SrsRtmpServer;
class SrsCommonMessage;
class SrsRtmpConn;
class SrsSource;
class SrsRequest;
class SrsConsumer;
class SrsHttpConn;
class SrsResponseOnlyHttpConn;

/**
 * for the recv thread to handle the message.
 */
class ISrsMessageHandler {
public:
    ISrsMessageHandler();
    virtual ~ISrsMessageHandler();
public:
    /**
    * whether the handler can handle,
    * for example, when queue recv handler got an message,
    * it wait the user to process it, then the recv thread
    * never recv message util the handler is ok.
    */
    virtual bool can_handle() = 0;
    /**
     * process the received message.
     * @remark user must free this message.
     */
    virtual int handle(SrsCommonMessage* msg) = 0;
    /**
    * when recv message error.
    */
    virtual void on_recv_error(int ret) = 0;
    /**
    * when thread start or stop,
    * for example, the message handler can set whether auto response.
    */
    virtual void on_thread_start() = 0;
    virtual void on_thread_stop() = 0;
};

/**
 * the recv thread, use message handler to handle each received message.
 */
class SrsRecvThread : public ISrsReusableThread2Handler {
protected:
    SrsReusableThread2* trd;
    ISrsMessageHandler* handler;
    SrsRtmpServer* rtmp;
    int timeout;
public:
    SrsRecvThread(ISrsMessageHandler* msg_handler, SrsRtmpServer* rtmp_sdk, int timeout_ms);
    virtual ~SrsRecvThread();
public:
    virtual int cid();
public:
    virtual int start();
    virtual void stop();
    virtual void stop_loop();
    // interface ISrsReusableThread2Handler
public:
    virtual int cycle();
    virtual void on_thread_start();
    virtual void on_thread_stop();
};

/**
* the recv thread used to replace the timeout recv,
* which hurt performance for the epoll_ctrl is frequently used.
* @see: SrsRtmpConn::playing
*/
class SrsQueueRecvThread : public ISrsMessageHandler {
private:
    std::vector<SrsCommonMessage*> queue;
    SrsRecvThread trd;
    SrsRtmpServer* rtmp;
    // the recv thread error code.
    int recv_error_code;
    SrsConsumer* _consumer;
public:
    SrsQueueRecvThread(SrsConsumer* consumer, SrsRtmpServer* rtmp_sdk, int timeout_ms);
    virtual ~SrsQueueRecvThread();
public:
    virtual int start();
    virtual void stop();
public:
    virtual bool empty();
    virtual int size();
    virtual SrsCommonMessage* pump();
    virtual int error_code();
public:
    virtual bool can_handle();
    virtual int handle(SrsCommonMessage* msg);
    virtual void on_recv_error(int ret);
public:
    virtual void on_thread_start();
    virtual void on_thread_stop();
};

/**
* the publish recv thread got message and callback the source method to process message.
*/
class SrsPublishRecvThread : virtual public ISrsMessageHandler
#ifdef SRS_PERF_MERGED_READ
    , virtual public IMergeReadHandler
#endif
    , virtual public ISrsReloadHandler {
private:
    SrsRecvThread trd;
    SrsRtmpServer* rtmp;
    SrsRequest* req;
    // the msgs already got.
    int64_t _nb_msgs;
    // The video frames we got.
    uint64_t video_frames;
    // for mr(merged read),
    bool mr;
    int mr_fd;
    int mr_sleep;
    // for realtime
    bool realtime;
    // the recv thread error code.
    int recv_error_code;
    SrsRtmpConn* _conn;
    // the params for conn callback.
    SrsSource* _source;
    bool _is_fmle;
    bool _is_edge;
    // the error timeout cond
    st_cond_t error;
    // merged context id.
    int cid;
    int ncid;
public:
    SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk,
                         SrsRequest* _req, int mr_sock_fd, int timeout_ms,
                         SrsRtmpConn* conn, SrsSource* source, bool is_fmle, bool is_edge);
    virtual ~SrsPublishRecvThread();
public:
    /**
    * wait for error for some timeout.
    */
    virtual int wait(int timeout_ms);
    virtual int64_t nb_msgs();
    virtual uint64_t nb_video_frames();
    virtual int error_code();
    virtual void set_cid(int v);
    virtual int get_cid();
public:
    virtual int start();
    virtual void stop();
    virtual void on_thread_start();
    virtual void on_thread_stop();
    // interface ISrsMessageHandler
public:
    virtual bool can_handle();
    virtual int handle(SrsCommonMessage* msg);
    virtual void on_recv_error(int ret);
    // interface IMergeReadHandler
public:
#ifdef SRS_PERF_MERGED_READ
    virtual void on_read(ssize_t nread);
#endif
    // interface ISrsReloadHandler
public:
    virtual int on_reload_vhost_mr(const std::string& vhost);
    virtual int on_reload_vhost_realtime(const std::string& vhost);
private:
    virtual void set_socket_buffer(int sleep_ms);
};

/**
 * The HTTP receive thread, try to read messages util EOF.
 * For example, the HTTP FLV serving thread will use the receive thread to break
 * when client closed the request, to avoid FD leak.
 */
class SrsHttpRecvThread : public ISrsOneCycleThreadHandler {
private:
    SrsResponseOnlyHttpConn* conn;
    SrsOneCycleThread* trd;
    int error;
public:
    SrsHttpRecvThread(SrsResponseOnlyHttpConn* c);
    virtual ~SrsHttpRecvThread();
public:
    virtual int start();
public:
    virtual int error_code();
    // interface ISrsOneCycleThreadHandler
public:
    virtual int cycle();
};
